package market

import (
	"encoding/json"
	"fmt"
	"log"
	"strings"
	"sync"
	"time"
)

type WSMonitor struct {
	wsClient        *WSClient
	combinedClient  *CombinedStreamsClient
	symbols         []string
	featuresMap     sync.Map
	alertsChan      chan Alert
	klineDataMap5m  sync.Map // 存储每个交易对的5分钟K线历史数据
	klineDataMap15m sync.Map // 存储每个交易对的15分钟K线历史数据
	klineDataMap1h  sync.Map // 存储每个交易对的1小时K线历史数据
	klineDataMap4h  sync.Map // 存储每个交易对的K线历史数据
	klineDataMap1d  sync.Map // 存储每个交易对的日线K线历史数据
	tickerDataMap   sync.Map // 存储每个交易对的ticker数据
	batchSize       int
	filterSymbols   sync.Map // 使用sync.Map来存储需要监控的币种和其状态
	symbolStats     sync.Map // 存储币种统计信息
	FilterSymbol    []string //经过筛选的币种
}

type SymbolStats struct {
	LastActiveTime   time.Time
	AlertCount       int
	VolumeSpikeCount int
	LastAlertTime    time.Time
	Score            float64 // 综合评分
}

// KlineCacheEntry 带时间戳的K线缓存条目
// 用于检测数据新鲜度，防止使用过期数据
type KlineCacheEntry struct {
	Klines     []Kline   // K线数据
	ReceivedAt time.Time // 数据接收时间
}

var WSMonitorCli *WSMonitor
var subKlineTime = []string{"5m", "15m", "1h", "4h", "1d"} // 管理订阅流的K线周期

func NewWSMonitor(batchSize int) *WSMonitor {
	WSMonitorCli = &WSMonitor{
		wsClient:       NewWSClient(),
		combinedClient: NewCombinedStreamsClient(batchSize),
		alertsChan:     make(chan Alert, 1000),
		batchSize:      batchSize,
	}
	return WSMonitorCli
}

func (m *WSMonitor) Initialize(coins []string) error {
	log.Println("初始化WebSocket监控器...")
	// 获取交易对信息
	apiClient := NewAPIClient()
	// 如果不指定交易对，则使用market市场的所有交易对币种
	if len(coins) == 0 {
		exchangeInfo, err := apiClient.GetExchangeInfo()
		if err != nil {
			return err
		}
		// 筛选永续合约交易对 --仅测试时使用
		//exchangeInfo.Symbols = exchangeInfo.Symbols[0:2]
		for _, symbol := range exchangeInfo.Symbols {
			if symbol.Status == "TRADING" && symbol.ContractType == "PERPETUAL" && strings.ToUpper(symbol.Symbol[len(symbol.Symbol)-4:]) == "USDT" {
				m.symbols = append(m.symbols, symbol.Symbol)
				m.filterSymbols.Store(symbol.Symbol, true)
			}
		}
	} else {
		m.symbols = coins
	}

	log.Printf("找到 %d 个交易对", len(m.symbols))
	// 初始化历史数据
	if err := m.initializeHistoricalData(); err != nil {
		log.Printf("初始化历史数据失败: %v", err)
	}

	return nil
}

func (m *WSMonitor) initializeHistoricalData() error {
	apiClient := NewAPIClient()

	var wg sync.WaitGroup
	semaphore := make(chan struct{}, 5) // 限制并发数

	for _, symbol := range m.symbols {
		wg.Add(1)
		semaphore <- struct{}{}

		go func(s string) {
			defer wg.Done()
			defer func() { <-semaphore }()

			// 获取历史K线数据 - 5m
			klines, err := apiClient.GetKlines(s, "5m", 100)
			if err != nil {
				log.Printf("获取 %s 历史数据失败: %v", s, err)
				return
			}
			if len(klines) > 0 {
				entry := &KlineCacheEntry{
					Klines:     klines,
					ReceivedAt: time.Now(),
				}
				m.klineDataMap5m.Store(s, entry)
				log.Printf("已加载 %s 的历史K线数据-5m: %d 条", s, len(klines))
			}

			// 获取历史K线数据 - 15m
			klines15m, err := apiClient.GetKlines(s, "15m", 100)
			if err != nil {
				log.Printf("获取 %s 历史数据失败: %v", s, err)
				return
			}
			if len(klines15m) > 0 {
				entry15m := &KlineCacheEntry{
					Klines:     klines15m,
					ReceivedAt: time.Now(),
				}
				m.klineDataMap15m.Store(s, entry15m)
				log.Printf("已加载 %s 的历史K线数据-15m: %d 条", s, len(klines15m))
			}

			// 获取历史K线数据 - 1h
			klines1h, err := apiClient.GetKlines(s, "1h", 100)
			if err != nil {
				log.Printf("获取 %s 历史数据失败: %v", s, err)
				return
			}
			if len(klines1h) > 0 {
				entry1h := &KlineCacheEntry{
					Klines:     klines1h,
					ReceivedAt: time.Now(),
				}
				m.klineDataMap1h.Store(s, entry1h)
				log.Printf("已加载 %s 的历史K线数据-1h: %d 条", s, len(klines1h))
			}

			// 获取历史K线数据 - 4h
			klines4h, err := apiClient.GetKlines(s, "4h", 100)
			if err != nil {
				log.Printf("获取 %s 历史数据失败: %v", s, err)
				return
			}
			if len(klines4h) > 0 {
				entry4h := &KlineCacheEntry{
					Klines:     klines4h,
					ReceivedAt: time.Now(),
				}
				m.klineDataMap4h.Store(s, entry4h)
				log.Printf("已加载 %s 的历史K线数据-4h: %d 条", s, len(klines4h))
			}

			// 获取历史K线数据 - 1d
			klines1d, err := apiClient.GetKlines(s, "1d", 100)
			if err != nil {
				log.Printf("获取 %s 历史数据失败: %v", s, err)
				return
			}
			if len(klines1d) > 0 {
				entry1d := &KlineCacheEntry{
					Klines:     klines1d,
					ReceivedAt: time.Now(),
				}
				m.klineDataMap1d.Store(s, entry1d)
				log.Printf("已加载 %s 的历史K线数据-1d: %d 条", s, len(klines1d))
			}
		}(symbol)
	}

	wg.Wait()
	return nil
}

func (m *WSMonitor) Start(coins []string) {
	log.Printf("启动WebSocket实时监控...")
	// 初始化交易对
	err := m.Initialize(coins)
	if err != nil {
		log.Printf("❌ 初始化币种失败: %v", err)
		return
	}

	err = m.combinedClient.Connect()
	if err != nil {
		log.Printf("❌ 批量订阅流失败: %v", err)
		return
	}
	// 订阅所有交易对
	err = m.subscribeAll()
	if err != nil {
		log.Printf("❌ 订阅币种交易对失败: %v", err)
		return
	}
}

// subscribeSymbol 注册监听
func (m *WSMonitor) subscribeSymbol(symbol, st string) []string {
	var streams []string
	stream := fmt.Sprintf("%s@kline_%s", strings.ToLower(symbol), st)
	ch := m.combinedClient.AddSubscriber(stream, 100)
	streams = append(streams, stream)
	go m.handleKlineData(symbol, ch, st)

	return streams
}
func (m *WSMonitor) subscribeAll() error {
	// 执行批量订阅
	log.Println("开始订阅所有交易对...")
	for _, symbol := range m.symbols {
		for _, st := range subKlineTime {
			m.subscribeSymbol(symbol, st)
		}
	}
	for _, st := range subKlineTime {
		err := m.combinedClient.BatchSubscribeKlines(m.symbols, st)
		if err != nil {
			log.Printf("❌ 订阅 %s K线失败: %v", st, err)
			return err
		}
	}
	log.Println("所有交易对订阅完成")
	return nil
}

func (m *WSMonitor) handleKlineData(symbol string, ch <-chan []byte, _time string) {
	for data := range ch {
		var klineData KlineWSData
		if err := json.Unmarshal(data, &klineData); err != nil {
			log.Printf("解析Kline数据失败: %v", err)
			continue
		}
		m.processKlineUpdate(symbol, klineData, _time)
	}
}

func (m *WSMonitor) getKlineDataMap(_time string) *sync.Map {
	var klineDataMap *sync.Map
	if _time == "5m" {
		klineDataMap = &m.klineDataMap5m
	} else if _time == "15m" {
		klineDataMap = &m.klineDataMap15m
	} else if _time == "1h" {
		klineDataMap = &m.klineDataMap1h
	} else if _time == "4h" {
		klineDataMap = &m.klineDataMap4h
	} else if _time == "1d" {
		klineDataMap = &m.klineDataMap1d
	} else {
		klineDataMap = &sync.Map{}
	}
	return klineDataMap
}
func (m *WSMonitor) processKlineUpdate(symbol string, wsData KlineWSData, _time string) {
	// 转换WebSocket数据为Kline结构
	kline := Kline{
		OpenTime:  wsData.Kline.StartTime,
		CloseTime: wsData.Kline.CloseTime,
		Trades:    wsData.Kline.NumberOfTrades,
	}
	kline.Open, _ = parseFloat(wsData.Kline.OpenPrice)
	kline.High, _ = parseFloat(wsData.Kline.HighPrice)
	kline.Low, _ = parseFloat(wsData.Kline.LowPrice)
	kline.Close, _ = parseFloat(wsData.Kline.ClosePrice)
	kline.Volume, _ = parseFloat(wsData.Kline.Volume)
	kline.High, _ = parseFloat(wsData.Kline.HighPrice)
	kline.QuoteVolume, _ = parseFloat(wsData.Kline.QuoteVolume)
	kline.TakerBuyBaseVolume, _ = parseFloat(wsData.Kline.TakerBuyBaseVolume)
	kline.TakerBuyQuoteVolume, _ = parseFloat(wsData.Kline.TakerBuyQuoteVolume)
	// 更新K线数据
	var klineDataMap = m.getKlineDataMap(_time)
	value, exists := klineDataMap.Load(symbol)
	var klines []Kline
	if exists {
		// 从缓存条目中提取K线数据
		entry := value.(*KlineCacheEntry)
		klines = entry.Klines

		// 检查是否是新的K线
		if len(klines) > 0 && klines[len(klines)-1].OpenTime == kline.OpenTime {
			// 更新当前K线
			klines[len(klines)-1] = kline
		} else {
			// 添加新K线
			klines = append(klines, kline)

			// 保持数据长度
			if len(klines) > 100 {
				klines = klines[1:]
			}
		}
	} else {
		klines = []Kline{kline}
	}

	// 存储时加上接收时间戳
	entry := &KlineCacheEntry{
		Klines:     klines,
		ReceivedAt: time.Now(),
	}
	klineDataMap.Store(symbol, entry)
}

func (m *WSMonitor) GetCurrentKlines(symbol string, duration string) ([]Kline, error) {
	// 对每一个进来的symbol检测是否存在内类 是否的话就订阅它
	value, exists := m.getKlineDataMap(duration).Load(symbol)
	if !exists {
		// 如果Ws数据未初始化完成时,单独使用api获取 - 兼容性代码 (防止在未初始化完成是,已经有交易员运行)
		apiClient := NewAPIClient()
		klines, err := apiClient.GetKlines(symbol, duration, 100)
		if err != nil {
			return nil, fmt.Errorf("获取%v分钟K线失败: %v", duration, err)
		}

		// 动态缓存进缓存（使用 KlineCacheEntry 包装，加上时间戳）
		entry := &KlineCacheEntry{
			Klines:     klines,
			ReceivedAt: time.Now(),
		}
		m.getKlineDataMap(duration).Store(strings.ToUpper(symbol), entry)

		// 订阅 WebSocket 流
		subStr := m.subscribeSymbol(symbol, duration)
		subErr := m.combinedClient.subscribeStreams(subStr)
		log.Printf("动态订阅流: %v", subStr)
		if subErr != nil {
			log.Printf("警告: 动态订阅%v分钟K线失败: %v (使用API数据)", duration, subErr)
		}

		// ✅ FIX: 返回深拷贝而非引用
		result := make([]Kline, len(klines))
		copy(result, klines)
		return result, nil
	}

	// 从缓存读取数据
	entry := value.(*KlineCacheEntry)

	// ✅ 检查数据新鲜度（防止使用过期数据）
	// 使用 15 分钟阈值：对于 5m 和 4h K线都适用
	// - 5m K线：15分钟 = 3个周期，足以检测 WebSocket 停止
	// - 4h K线：虽然新 K线 4小时才生成，但当前 K线 是实时更新的
	dataAge := time.Since(entry.ReceivedAt)
	maxAge := 15 * time.Minute

	if dataAge > maxAge {
		// 数据过期，返回错误（不 fallback API，避免增加负担）
		// 这表明 WebSocket 可能未正常工作，需要修复根本原因
		return nil, fmt.Errorf("%s 的 %s K线数据已过期 (%.1f 分钟)，WebSocket 可能未正常工作",
			symbol, duration, dataAge.Minutes())
	}

	// 数据新鲜，返回缓存数据（深拷贝）
	klines := entry.Klines
	result := make([]Kline, len(klines))
	copy(result, klines)
	return result, nil
}

func (m *WSMonitor) Close() {
	m.wsClient.Close()
	close(m.alertsChan)
}
